package com.youbu.demo;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

/**
 * @author sunfangwei
 **/
public class WordCountDemo {

	public static void main(String[] args) {
		String filePath = args[0];
		String resultPath = args[1];
		SparkConf conf = new SparkConf().setAppName("YoubuWordCount").setMaster("spark://ServerMaster:7077");
		JavaSparkContext sc = new JavaSparkContext(conf);
		JavaRDD<String> input = sc.textFile(filePath);
		JavaRDD<String> words =input.flatMap(new FlatMapFunction<String, String>() {
			@Override
			public Iterator<String> call(String s) throws Exception {
				return Arrays.asList(s.split(" ")).iterator();
			}
		});
		JavaPairRDD<String,Integer> counts = words.mapToPair(
				new PairFunction<String, String, Integer>() {
					@Override
					public Tuple2<String, Integer> call(String s) throws Exception {
						return new Tuple2<>(s,1);
					}
				}
		).reduceByKey(new Function2<Integer, Integer, Integer>() {
			@Override
			public Integer call(Integer integer, Integer integer2) throws Exception {
				return integer2+integer;
			}
		});
		counts.saveAsTextFile(resultPath+"/"+System.currentTimeMillis());
	}

}
